package com.learn.kafka.service;


import com.learn.kafka.config.KafkaConfigProperties;
import com.learn.kafka.domain.Order;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.stream.Collectors;

/**
 * 多写服务类（核心逻辑）
 * @author:         MI
 * @email:          448341911@qq.com
 * @createTime:     2025/8/9 10:39
 * @updateUser:     MI
 * @updateTime:     2025/8/9 10:39
 * @updateRemark:   修改内容
 * @version:        1.0
 */
@Service
public class OrderMultiWriteService {


    private final KafkaTemplate<String, Object> kafkaTemplate;
    private final List<String> targetTopics;  // 多写目标 Topic 列表

    // 注入 KafkaTemplate 和多写配置
    public OrderMultiWriteService(KafkaTemplate<String, Object> kafkaTemplate,
                                  KafkaConfigProperties kafkaConfigProperties) {
        this.kafkaTemplate = kafkaTemplate;
        this.targetTopics = kafkaConfigProperties.getTopics();
    }

    /**
     * 多写订单数据到所有目标 Topic（带事务支持）
     */
    @Transactional  // 事务注解：确保所有 Topic 要么全成功，要么全失败
    public void multiWriteOrder(Order order) {
        // 遍历目标 Topic，逐个发送（可优化为并行发送）
        List<CompletableFuture<SendResult<String, Object>>> futures = targetTopics.stream()
                .map(topic -> kafkaTemplate.send(topic, order.getOrderId(), order))  // 以 orderId 为 key，保证同订单分区一致
                .collect(Collectors.toList());

        // 等待所有发送完成（同步阻塞，也可异步处理结果）
        futures.forEach(future -> {
            try {
                future.get();  // 阻塞获取结果，若失败会抛出异常，触发事务回滚
                System.out.printf("订单 %s 成功写入 Topic: %s%n", order.getOrderId(), future.get().getRecordMetadata().topic());
            } catch (Exception e) {
                throw new RuntimeException("订单多写失败", e);  // 抛出异常，触发事务回滚
            }
        });
    }
}
